package yuekao6.ods;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import yuekao6.util.BrodcastProcess;
import yuekao6.util.KafkaUtil;
import yuekao6.util.MyDeserializationSchemaFunction;

public class Broadcast {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//（2）获取流处理环境，读取主流数据，主流数据结构转换，连接主流和配置流，并在获取配置流后创建维度表，处理广播流，将数据写出到HBase维表（5分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSources("tms_ods"));

        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop-single")
                .port(3306)
                .databaseList("yuekao06") // monitor all tables under inventory database
                .startupOptions(StartupOptions.initial())//全量
                .username("root")
                .password("root")
                .deserializer(new MyDeserializationSchemaFunction()) // converts SourceRecord to String
                .build();

        DataStreamSource<String> streamSource1 = env.addSource(sourceFunction);

//        streamSource.print();
//        streamSource1.print();

        //广播流
        MapStateDescriptor test = new MapStateDescriptor("test", String.class, String.class);

        BroadcastStream<String> broadcast = streamSource1.broadcast(test);

        //连接主流和配置流
        BroadcastConnectedStream<String, String> connect = streamSource.connect(broadcast);


        connect.process(new BrodcastProcess(test));

        env.execute();
    }
}
